package com.foreveross.fetl.common.db.hbase;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;



@Component
public class HbaseManager {
	
	protected Log logger= LogFactory.getLog(getClass());
	
	private HbaseDriverManager driverManager;
	
	public HbaseManager(){
			driverManager = HbaseDriverManager.getInstance();
	}
	
	 /**
     * 插入一行记录
     */
    public void addRecord(String tableName, HBaseRowModel model)
            throws Exception{
    	HTableInterface table = null;
    	Put put = null;
        try {
        	if(StringUtils.isNotBlank(tableName) &&
        			model != null){
        	 	put = new Put(Bytes.toBytes(model.getRowkey()));
        	 	//循环将所有值都放入到put中
        		for(HBaseCellModel cell : model.getCells()){
        			if(StringUtils.isNotBlank(cell.getValue())){//写还是不写
        				put.add(Bytes.toBytes(cell.getFamily()),Bytes.toBytes(cell.getQualifier()), Bytes.toBytes(cell.getValue()));
        			}
        		}
	            table = driverManager.getHtable(tableName);
	            table.put(put);
        	}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、model为空，没有插入数据!：");
            	}
        	}
        } finally{
        	driverManager.releaseHTable(tableName);
        	table = null;
        	put = null;
        }
    }
    /**
     * 插入单个列记录
     */
    public void addRecord(String tableName,HBaseCellModel cell)
            throws Exception{
    	
    	if(StringUtils.isBlank(cell.getValue())) return;
    	 
    	HTableInterface table = null;
    	Put put = null;
        try {
        	if(StringUtils.isNotBlank(tableName) &&
        			cell != null && cell.getRowModel() != null){
	            put = new Put(Bytes.toBytes(cell.getRowModel().getRowkey()));
	           
	            	put.add(Bytes.toBytes(cell.getFamily()),Bytes.toBytes(cell.getQualifier()), Bytes.toBytes(cell.getValue()));
	            
	            table = driverManager.getHtable(tableName);
	            table.put(put);
        	}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、cell为空，没有插入数据!：");
            	}
        	}
        } finally{
        	driverManager.releaseHTable(tableName);
        	table = null;
        	put = null;
        }
    }
 
    /**
     * 删除一行记录
     */
    public void delOneRecord (String tableName, String rowKey) throws Exception{
    	HTableInterface table = null;
    	try{
    		if(StringUtils.isNotBlank(tableName) && StringUtils.isNotBlank(rowKey)){
    			table = driverManager.getHtable(tableName);
    			table.delete(new Delete(rowKey.getBytes()));
    		}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、rowkey为空，没有删除数据!：");
            	}
        	}
    	} finally{
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    }
    
    /**
     * 删除多行记录
     */
    public void delMultRecord (String tableName, String[] rowKeys) throws Exception{
    	HTableInterface table = null;
    	List<Delete> deletes = null;
    	try{
    		if(StringUtils.isNotBlank(tableName) && rowKeys != null && rowKeys.length > 0){
    			table = driverManager.getHtable(tableName);
    			deletes = new ArrayList<Delete>();
    			for(String k : rowKeys){
    				deletes.add(new Delete(k.getBytes()));
    			}
    			
    			table.delete(deletes);
    		}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、rowKeys为空，没有删除数据!：");
            	}
        	}
    	} finally{
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    }
     
    /**
     * 查找一行记录
     * @throws Exception 
     */
    public HBaseRowModel getOneRecord(String tableName, String rowKey) throws Exception{
    	HTableInterface table = null;
    	Get get = null;
    	Result rs = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
    	try{
    		model = new HBaseRowModel();
    		if(StringUtils.isNotBlank(tableName) && StringUtils.isNotBlank(rowKey)){
    			table = driverManager.getHtable(tableName);
    			get = new Get(rowKey.getBytes());
    			rs = table.get(get);
    			
    			model.setRowkey(rowKey);
    			
    			for(KeyValue kv : rs.raw()){
    				cell = new HBaseCellModel();
    				cell.setFamily(new String(kv.getFamily()));
    				cell.setQualifier(new String(kv.getQualifier()));
    				cell.setTimestamp(kv.getTimestamp());
    				cell.setValue(new String(kv.getValue()));
    				
    				model.addCell(cell);
    		   }
    		}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、rowkey为空，没有删除数据!：");
            	}
        	}
    	} finally{
    		rs = null;
    		get = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return model;
    }
     
    /**
     * 查询多行记录
     * @throws Exception 
     */
    public List<HBaseRowModel> getMultRecord(String tableName, Filter filter) throws Exception {
    	List<HBaseRowModel> results = null;
    	HTableInterface table = null;
    	Scan s = null;
    	ResultScanner ss = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
        try{
        	results = new ArrayList<HBaseRowModel>();
              s = new Scan();
              
             if(filter != null) s.setFilter(filter);
             
             table = driverManager.getHtable(tableName);
             ss = table.getScanner(s);
             for(Result r:ss){
            	 //行
            	 model = new HBaseRowModel(new String(r.getRow()));
                 for(KeyValue kv : r.raw()){
                	 //单元格
                	cell = new HBaseCellModel();
                	cell.setFamily(new String(kv.getFamily()));
     				cell.setQualifier(new String(kv.getQualifier()));
     				cell.setTimestamp(kv.getTimestamp());
     				cell.setValue(new String(kv.getValue()));
     				
     				model.addCell(cell);
                 }
                 
                 results.add(model);
             }
            
        } finally{
    		s = null;
    		ss = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return results;
    }
    
    /**
     * 查询多行记录
     * @throws Exception 
     */
    public List<HBaseRowModel> getMultRecord(
    		String tableName, Filter filter,
    		String start,String end) throws Exception {
    	List<HBaseRowModel> results = null;
    	HTableInterface table = null;
    	Scan s = null;
    	ResultScanner ss = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
        try{
			results =  new ArrayList<HBaseRowModel>();
			s = new Scan();
			if (StringUtils.isNotBlank(start))
				s.setStartRow(start.getBytes());
			if (StringUtils.isNotBlank(end))
				s.setStopRow(end.getBytes());
			if (filter != null)
				s.setFilter(filter);
			table = driverManager.getHtable(tableName);
			ss = table.getScanner(s);
			for (Result r : ss) {
				// 行
				model = new HBaseRowModel(new String(r.getRow()));
				for (KeyValue kv : r.raw()) {
					// 单元格
					cell = new HBaseCellModel();
					cell.setFamily(new String(kv.getFamily()));
					cell.setQualifier(new String(kv.getQualifier()));
					cell.setTimestamp(kv.getTimestamp());
					cell.setValue(new String(kv.getValue()));

					model.addCell(cell);
				}
				results.add(model);
			}
        } finally{
    		s = null;
    		ss = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return results;
    }
    
    
    /**
	 * 生成航班数据的hbase主键
	 * 抓取日期(年月日时)_渠道_区域ID_起点(使用三字码)_终点(使用三字码)_起飞时间(年月日时)_航班ID_5位随即数
     *	10位 +    		  5位 +2位 + 3位   +        	3位      + 				10位+   		  8位 +    		5位
	 */
    
/*    public static void main(String[] arg) throws Exception{
    	//logger.info("抓取引警系统配置文件统一注册......");
    	ConfigContainer.getInstance().register(HBasePropertiesLoader.class);
    	List<String> rowkeys = new ArrayList<String>();
    //	rowkeys.add("crawl10.108.1.230_1_CrawlScheduler_Worker-28");
    	rowkeys.add("20130530_20006_03_KMG_PEK_201306120000000");
    	
    	
    	
    	HbaseManager manager = new HbaseManager();
    	int i = 1;
    	for(String rowkey : rowkeys){
    		 HBaseRowModel model = manager.getOneRecord("CRAWL_PF_SOURCE_DATA", rowkey);
        	 HBaseCellModel cell = model.lookupCell("CONTENTS", "CONTENTS");
        	 org.apache.commons.io.FileUtils.writeStringToFile(new File("F:/temp/hbase/source-"+(i++)+".txt"), 
        			 new String(cell.getValue().getBytes(),"gb2312"), false);
    	}
    	 //System.out.println(cell.getValue());
    }*/
}
